Node / meshtastic / Meshtastic-Android / files / core / takserver / src / commonMain / kotlin / org / meshtastic / core / takserver / MeshToCotBroadcaster.kt

Displaying Raw • Download

T8b949e/*

* Copyright (c) 2026 Meshtastic LLC

*

* This program is free software: you can redistribute it and/or modify

* it under the terms of the GNU General Public License as published by

* the Free Software Foundation, either version 3 of the License, or

* (at your option) any later version.

*

* This program is distributed in the hope that it will be useful,

* but WITHOUT ANY WARRANTY; without even the implied warranty of

* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the

* GNU General Public License for more details.

*

* You should have received a copy of the GNU General Public License

* along with this program. If not, see <https://www.gnu.org/licenses/>.

*/

Tff7b72package T7ee787org.meshtastic.core.takserver

Tff7b72import T7ee787co.touchlab.kermit.Logger

Tff7b72import T7ee787kotlinx.coroutines.CoroutineScope

Tff7b72import T7ee787kotlinx.coroutines.Job

Tff7b72import T7ee787kotlinx.coroutines.coroutineScope

Tff7b72import T7ee787kotlinx.coroutines.delay

Tff7b72import T7ee787kotlinx.coroutines.flow.collectLatest

Tff7b72import T7ee787kotlinx.coroutines.launch

Tff7b72import T7ee787kotlinx.coroutines.sync.Mutex

Tff7b72import T7ee787kotlinx.coroutines.sync.withLock

Tff7b72import T7ee787org.meshtastic.core.di.CoroutineDispatchers

Tff7b72import T7ee787org.meshtastic.core.model.Node

Tff7b72import T7ee787org.meshtastic.core.repository.NodeRepository

Tff7b72import T7ee787org.meshtastic.core.repository.TakPrefs

Tff7b72import T7ee787kotlin.concurrent.Volatile

Tff7b72import T7ee787kotlin.concurrent.atomics.AtomicBoolean

Tff7b72import T7ee787kotlin.concurrent.atomics.ExperimentalAtomicApi

Tff7b72import T7ee787kotlin.time.Instant

T8b949e/**

* Publishes Meshtastic node-database entries to connected TAK clients as CoT contacts.

*

* This is the mesh -> ATAK visualization path, and it is entirely local: synthesized node CoT travels app -> TAK client

* and never crosses the mesh. Mesh-originated TAKPacket traffic is [TAKMeshIntegration]'s job; the two are independent.

*

* Opt-in via [TakPrefs.isMeshToCotEnabled] and additionally gated on the TAK server running, because the owning

* [TAKMeshIntegration] is itself only started while the server is enabled.

*/

Tf0883e@OptInTb4b4b4(Te6edf3ExperimentalAtomicApiTff7b72::Te6edf3classTb4b4b4)

Tff7b72class T56d364MeshToCotBroadcasterTb4b4b4(

Tff7b72private Tff7b72val Te6edf3takServerManagerTb4b4b4: Te6edf3TAKServerManagerTb4b4b4,

Tff7b72private Tff7b72val Te6edf3nodeRepositoryTb4b4b4: Te6edf3NodeRepositoryTb4b4b4,

Tff7b72private Tff7b72val Te6edf3takPrefsTb4b4b4: Te6edf3TakPrefsTb4b4b4,

Tff7b72private Tff7b72val Te6edf3dispatchersTb4b4b4: Te6edf3CoroutineDispatchersTb4b4b4,

Tb4b4b4) Tb4b4b4{

Tff7b72private Tff7b72val Te6edf3isRunning Tff7b72= Te6edf3AtomicBooleanTb4b4b4(Tff7b72falseTb4b4b4)

Tf0883e@Volatile Tff7b72private Tff7b72var Te6edf3jobTb4b4b4: Te6edf3Job? Tff7b72= Tff7b72null

T8b949e// Last CoT sent per node number, time-normalized (see normalizeForDedup) so an unchanged node

T8b949e// is not re-sent on every unrelated nodeDB emission. Guarded by sentMutex.

Tff7b72private Tff7b72val Te6edf3lastSent Tff7b72= Te6edf3mutableMapOfTff7b72<Tffa657IntTb4b4b4, Te6edf3CoTMessageTff7b72>Tb4b4b4(Tb4b4b4)

Tff7b72private Tff7b72val Te6edf3sentMutex Tff7b72= Te6edf3MutexTb4b4b4(Tb4b4b4)

Tff7b72fun Td2a8ffstartTb4b4b4(Te6edf3scopeTb4b4b4: Te6edf3CoroutineScopeTb4b4b4) Tb4b4b4{

T8b949e// CAS, not a job-null check: two concurrent start() calls must not both launch, and a

T8b949e// dead job left by a cancelled scope must not block every future start().

Tff7b72if Tb4b4b4(Tff7b72!Te6edf3isRunningTb4b4b4.Te6edf3compareAndSetTb4b4b4(Te6edf3expectedValue Tff7b72= Tff7b72falseTb4b4b4, Te6edf3newValue Tff7b72= Tff7b72trueTb4b4b4)Tb4b4b4) Tff7b72return

Te6edf3job Tff7b72=

Te6edf3scopeTb4b4b4.Te6edf3launchTb4b4b4(Te6edf3dispatchersTb4b4b4.Te6edf3defaultTb4b4b4) Tb4b4b4{

Tff7b72try Tb4b4b4{

Te6edf3takPrefsTb4b4b4.Te6edf3isMeshToCotEnabledTb4b4b4.Te6edf3collectLatest Tb4b4b4{ Te6edf3enabled Tff7b72-Tff7b72>

Tff7b72if Tb4b4b4(Tff7b72!Te6edf3enabledTb4b4b4) Tff7b72returnTf0883e@collectLatest

Te6edf3LoggerTb4b4b4.Te6edf3i Tb4b4b4{ Ta5d6ff"Ta5d6ffMesh-to-CoT enabled — publishing node contacts to TAK clientsTa5d6ff" Tb4b4b4}

Te6edf3runEnabledTb4b4b4(Tb4b4b4)

Tb4b4b4}

Tb4b4b4} Tff7b72finally Tb4b4b4{

T8b949e// Owning scope cancelled without stop(): release the guard so a later start()

T8b949e// on a fresh scope isn't refused forever.

Te6edf3isRunningTb4b4b4.Te6edf3storeTb4b4b4(Tff7b72falseTb4b4b4)

Tb4b4b4}

Tb4b4b4}

Tb4b4b4}

Tff7b72fun Td2a8ffstopTb4b4b4(Tb4b4b4) Tb4b4b4{

Tff7b72if Tb4b4b4(Tff7b72!Te6edf3isRunningTb4b4b4.Te6edf3compareAndSetTb4b4b4(Te6edf3expectedValue Tff7b72= Tff7b72trueTb4b4b4, Te6edf3newValue Tff7b72= Tff7b72falseTb4b4b4)Tb4b4b4) Tff7b72return

T8b949e// lastSent is deliberately NOT cleared here: cancel() doesn't join, so a still-finishing

T8b949e// publish() on another thread may hold sentMutex, and clearing unsynchronized would race

T8b949e// it. runEnabled() drops the state on the next enable instead.

Te6edf3jobTff7b72?.Te6edf3cancelTb4b4b4(Tb4b4b4)

Te6edf3job Tff7b72= Tff7b72null

Te6edf3LoggerTb4b4b4.Te6edf3i Tb4b4b4{ Ta5d6ff"Ta5d6ffMesh-to-CoT stoppedTa5d6ff" Tb4b4b4}

Tb4b4b4}

Tff7b72private Tff7b72suspend Tff7b72fun Td2a8ffrunEnabledTb4b4b4(Tb4b4b4) Tff7b72= Te6edf3coroutineScope Tb4b4b4{

T8b949e// Fresh dedup state per enable cycle, so nothing carried over from a previous run (or a

T8b949e// previous device) suppresses the initial publish.

Te6edf3clearSentTb4b4b4(Tb4b4b4)

T8b949e// A newly attached client has none of our prior broadcasts, so drop the dedup state

T8b949e// and replay every eligible node. Uses the live snapshot rather than the nodeDBbyNum

T8b949e// cache, which can briefly hold the previous transport's map after a device switch.

Te6edf3launch Tb4b4b4{

Te6edf3takServerManagerTb4b4b4.Te6edf3clientConnectedTb4b4b4.Te6edf3collect Tb4b4b4{

Te6edf3clearSentTb4b4b4(Tb4b4b4)

Te6edf3publishTb4b4b4(Te6edf3nodeRepositoryTb4b4b4.Te6edf3getNodeDbSnapshotTb4b4b4(Tb4b4b4)Tb4b4b4.Te6edf3valuesTb4b4b4, Te6edf3reason Tff7b72= Ta5d6ff"Ta5d6ffclient connectedTa5d6ff"Tb4b4b4)

Tb4b4b4}

Tb4b4b4}

T8b949e// Stationary nodes never change, so dedup alone would let their ATAK markers expire at

T8b949e// MESH_NODE_STALE_MINUTES. Re-send everything well inside that window.

Te6edf3launch Tb4b4b4{

Tff7b72while Tb4b4b4(Tff7b72trueTb4b4b4) Tb4b4b4{

Te6edf3delayTb4b4b4(Te6edf3MESH_TO_COT_REFRESH_INTERVAL_MSTb4b4b4)

Te6edf3clearSentTb4b4b4(Tb4b4b4)

Te6edf3publishTb4b4b4(Te6edf3nodeRepositoryTb4b4b4.Te6edf3getNodeDbSnapshotTb4b4b4(Tb4b4b4)Tb4b4b4.Te6edf3valuesTb4b4b4, Te6edf3reason Tff7b72= Ta5d6ff"Ta5d6ffperiodic refreshTa5d6ff"Tb4b4b4)

Tb4b4b4}

Tb4b4b4}

Te6edf3launch Tb4b4b4{ Te6edf3nodeRepositoryTb4b4b4.Te6edf3nodeDBbyNumTb4b4b4.Te6edf3collect Tb4b4b4{ Te6edf3publishTb4b4b4(Tffa657itTb4b4b4.Te6edf3valuesTb4b4b4, Te6edf3reason Tff7b72= Ta5d6ff"Ta5d6ffnode updateTa5d6ff"Tb4b4b4) Tb4b4b4} Tb4b4b4}

Tb4b4b4}

Tff7b72private Tff7b72suspend Tff7b72fun Td2a8ffpublishTb4b4b4(Te6edf3nodesTb4b4b4: Te6edf3CollectionTff7b72<Te6edf3NodeTff7b72>Tb4b4b4, Te6edf3reasonTb4b4b4: Tffa657StringTb4b4b4) Tb4b4b4{

T8b949e// Without a connected client every broadcast() would land in TAKServerManager's 50-entry

T8b949e// offline queue and evict genuine mesh CoT. Nothing is lost by skipping: connecting a

T8b949e// client triggers a full replay. Re-checked per node because a client can disconnect

T8b949e// mid-replay, and the inter-message spacing makes a large replay take whole seconds.

Tff7b72val Te6edf3ourNodeNum Tff7b72= Te6edf3nodeRepositoryTb4b4b4.Te6edf3myNodeInfoTb4b4b4.Te6edf3valueTff7b72?.Te6edf3myNodeNum

Tff7b72val Te6edf3eligible Tff7b72= Te6edf3nodesTb4b4b4.Te6edf3filter Tb4b4b4{ Tffa657itTb4b4b4.Te6edf3isEligibleForCotTb4b4b4(Te6edf3ourNodeNumTb4b4b4) Tb4b4b4}

Tff7b72var Te6edf3sent Tff7b72= T79c0ff0

Tff7b72for Tb4b4b4(Te6edf3node Tff7b72in Te6edf3eligibleTb4b4b4) Tb4b4b4{

Tff7b72if Tb4b4b4(Te6edf3takServerManagerTb4b4b4.Te6edf3connectionCountTb4b4b4.Te6edf3value Tff7b72<Tff7b72= T79c0ff0Tb4b4b4) Tff7b72return

Tff7b72val Te6edf3cot Tff7b72= Te6edf3nodeTb4b4b4.Te6edf3toCoTMessageTb4b4b4(Tb4b4b4)

Tff7b72if Tb4b4b4(Tff7b72!Te6edf3admitTb4b4b4(Te6edf3nodeTb4b4b4.Te6edf3numTb4b4b4, Te6edf3cotTb4b4b4)Tb4b4b4) Tff7b72continue

Te6edf3takServerManagerTb4b4b4.Te6edf3broadcastTb4b4b4(Te6edf3cotTb4b4b4)

Te6edf3sentTff7b72+Tff7b72+

Te6edf3delayTb4b4b4(Te6edf3MESH_TO_COT_BROADCAST_SPACING_MSTb4b4b4)

Tb4b4b4}

Tff7b72if Tb4b4b4(Te6edf3sent Tff7b72> T79c0ff0Tb4b4b4) Tb4b4b4{

Te6edf3LoggerTb4b4b4.Te6edf3i Tb4b4b4{ Ta5d6ff"Ta5d6ffMesh-to-CoT: sent Tffd700$Te6edf3sentTa5d6ff of Tffd700${Te6edf3eligibleTb4b4b4.Te6edf3sizeTffd700}Ta5d6ff eligible node(s) to TAK clients (Tffd700$Te6edf3reasonTa5d6ff)Ta5d6ff" Tb4b4b4}

Tb4b4b4}

Tb4b4b4}

T8b949e/** Records [cot] as the latest for [nodeNum], returning false when an identical event was already sent. */

Tff7b72private Tff7b72suspend Tff7b72fun Td2a8ffadmitTb4b4b4(Te6edf3nodeNumTb4b4b4: Tffa657IntTb4b4b4, Te6edf3cotTb4b4b4: Te6edf3CoTMessageTb4b4b4)Tb4b4b4: Tffa657Boolean Tff7b72= Te6edf3sentMutexTb4b4b4.Te6edf3withLock Tb4b4b4{

Tff7b72val Te6edf3normalized Tff7b72= Te6edf3normalizeForDedupTb4b4b4(Te6edf3cotTb4b4b4)

Tff7b72if Tb4b4b4(Te6edf3lastSentTff7b72[Te6edf3nodeNumTff7b72] Tff7b72=Tff7b72= Te6edf3normalizedTb4b4b4) Tff7b72return Tff7b72false

Te6edf3lastSentTff7b72[Te6edf3nodeNumTff7b72] Tff7b72= Te6edf3normalized

Tff7b72true

Tb4b4b4}

Tff7b72private Tff7b72suspend Tff7b72fun Td2a8ffclearSentTb4b4b4(Tb4b4b4) Tff7b72= Te6edf3sentMutexTb4b4b4.Te6edf3withLock Tb4b4b4{ Te6edf3lastSentTb4b4b4.Te6edf3clearTb4b4b4(Tb4b4b4) Tb4b4b4}

Tff7b72private Tff7b72companion Tff7b72object Tb4b4b4{

T8b949e/**

* Strips the timestamps so two events describing the same node state compare equal. Every other CoT field

* participates, so any change to position, callsign, battery, or telemetry re-broadcasts.

*/

Tff7b72fun Td2a8ffnormalizeForDedupTb4b4b4(Te6edf3cotTb4b4b4: Te6edf3CoTMessageTb4b4b4)Tb4b4b4: Te6edf3CoTMessage Tff7b72=

Te6edf3cotTb4b4b4.Te6edf3copyTb4b4b4(Te6edf3time Tff7b72= Te6edf3InstantTb4b4b4.Te6edf3DISTANT_PASTTb4b4b4, Te6edf3start Tff7b72= Te6edf3InstantTb4b4b4.Te6edf3DISTANT_PASTTb4b4b4, Te6edf3stale Tff7b72= Te6edf3InstantTb4b4b4.Te6edf3DISTANT_PASTTb4b4b4)

Tb4b4b4}

Tb4b4b4}